//
// Created by xuchao on 16-11-18.
//

#include "OkPgConnection.h"


using namespace OKPg;

OkPgConnection::OkPgConnection(const URL &url) : url_(url) {
    conn_ = PQconnectdb(url.toString().c_str());
    assert(PQstatus(conn_) == CONNECTION_OK);
    LOG(INFO) << "Connection create success!";
}

OkPgConnection::~OkPgConnection() {
    LOG(INFO) << "Connection destoryed success!";
    PQfinish(conn_);
}

OkPgConnection::OkPgConnection(OkPgConnection &&con) : url_(con.url_), conn_(std::move(con.conn_)) {
    PQclear(result_);
}

OkPgConnection &OkPgConnection::operator=(OkPgConnection &&con) {
    if (this == &con) {
        return *this;
    }
    if (result_) {
        PQclear(result_);
    }
    url_ = con.url_;
    conn_ = std::move(conn_);
    return *this;
}

static inline bool checkPQresultStatus(ExecStatusType status) {

    return status == PGRES_EMPTY_QUERY || status == PGRES_COMMAND_OK || status == PGRES_TUPLES_OK;

}

static void setPreparedStatement(const OkPgArgs &args, OkPgPreparedStatement *p) {
    int cuV = 0;
    const std::vector<std::string> argNameList = args.getArgNameList();
    for (std::vector<std::string>::const_iterator aci = argNameList.begin();
         aci != argNameList.end();
         ++aci) {
        OKPg::ArgValueType dt = args.getArgValueType(*aci);
        switch (dt) {
            case OKPg::STRING:
                (*p).setString(++cuV, args.getArgValueStr(*aci).c_str());
                break;
            case OKPg::INTEGER:
                (*p).setInt(++cuV, args.getArgValueInt(*aci));
                break;
            default:
                throw OkPgException("UNKNOWN TYPE!");
        }
    }
}

OkPgPreparedStatement OkPgConnection::newPreparedStatement(const OkPgArgs &args) {
    const int paramCount = args.getArgCount();

    OkPgPreparedStatement p(paramCount);
    //  LOG(INFO)<< args.toSqlString();
    result_ = PQprepare(conn_, p.stmtName_.c_str(), args.toSqlString().c_str(), 0, NULL);

    if (!checkPQresultStatus(PQresultStatus(result_))) {
        PQclear(result_);
        throw OkPgException(PQerrorMessage(conn_));
    }
    setPreparedStatement(args, &p);
    return p;
}

OkPgConnection *OkPgConnection::connNew(const URL &url) {
    return new OkPgConnection(url);
}

void OkPgConnection::connFree() {
    delete this;
}


bool OkPgConnection::ping() {
    return (PQstatus(conn_) == CONNECTION_OK);
}

void OkPgConnection::beginTransaction() {
    result_ = PQexec(conn_, "BEGIN TRANSACTION");
    if (!checkPQresultStatus(PQresultStatus(result_))) {
        PQclear(result_);
        throw OkPgException(PQerrorMessage(conn_));
    }
}

void OkPgConnection::commit() {
    result_ = PQexec(conn_, "END TRANSACTION");
    if (!checkPQresultStatus(PQresultStatus(result_))) {
        PQclear(result_);
        throw OkPgException(PQerrorMessage(conn_));
    }

}

void OkPgConnection::rollBack() {
    result_ = PQexec(conn_, "ROLLBACK TRANSACTION");
    if (!checkPQresultStatus(PQresultStatus(result_))) {
        PQclear(result_);
        throw OkPgException(PQerrorMessage(conn_));
    }
}

OkPgResultFuture OkPgConnection::execute(const string &cmd) {
    std::shared_ptr<OkPgResultPromise> promise(new folly::Promise<OkPgResultSet>);
    OkPgResultFuture furture = promise->getFuture();
    std::thread th([=] {
        try {
            result_ = PQexec(conn_, cmd.c_str());
            if (!checkPQresultStatus(PQresultStatus(result_))) {
                OkPgResultSet res(PQerrorMessage(conn_));
                PQclear(result_);
                promise->setValue(res);
            } else {
                OkPgResultSet res(string2int(PQcmdTuples(result_)));
                PQclear(result_);
                promise->setValue(res);
            }
        } catch (const OkPgException &ex) {
            OkPgResultSet res(ex.message_);
            promise->setValue(res);
        }
    });
    th.detach();
    return furture;
}

OkPgResultFuture OkPgConnection::execute(const OkPgArgs &args) {
    std::shared_ptr<OkPgResultPromise> promise(new folly::Promise<OkPgResultSet>);
    OkPgResultFuture furture = promise->getFuture();
    std::thread th([=] {
        try {
            beginTransaction();
            OkPgPreparedStatement p = newPreparedStatement(args);
            result_ = PQexecPrepared(conn_, p.stmtName_.c_str(), p.paramCount_, (const char **) p.paramValues_,
                                     p.paramLengths_,
                                     p.paramFormats_, 0);
            if (!checkPQresultStatus(PQresultStatus(result_))) {
                OkPgResultSet res(PQerrorMessage(conn_));
                rollBack();
                PQclear(result_);
                promise->setValue(res);
            } else {
                OkPgResultSet res(string2int(PQcmdTuples(result_)));
                commit();
                PQclear(result_);
                promise->setValue(res);
            }

        } catch (const OkPgException &ex) {
            rollBack();
            OkPgResultSet res(ex.message_);
            promise->setValue(res);
        }

    });
    th.detach();
    return furture;
}

OkPgResultFuture OkPgConnection::executeQuery(const string &cmd) {
    std::shared_ptr<OkPgResultPromise> promise(new folly::Promise<OkPgResultSet>);
    OkPgResultFuture future = promise->getFuture();
    std::thread th([=] {
        try {
            result_ = PQexec(conn_, cmd.c_str());
            if (!checkPQresultStatus(PQresultStatus(result_))) {
                OkPgResultSet res(PQerrorMessage(conn_));
                PQclear(result_);
                promise->setValue(res);
            } else {
                OkPgResultSet res(result_);
                PQclear(result_);
                promise->setValue(res);
            }
        } catch (const OkPgException &ex) {
            OkPgResultSet res(ex.message_);
            promise->setValue(res);
        }
    });
    th.detach();
    return future;
}


OkPgResultFuture OkPgConnection::executeQuery(const OkPgArgs &args) {
    std::shared_ptr<OkPgResultPromise> promise(new folly::Promise<OkPgResultSet>);
    OkPgResultFuture furture = promise->getFuture();
    std::thread th([=] {
        try {
            beginTransaction();
            OkPgPreparedStatement p = newPreparedStatement(args);
            result_ = PQexecPrepared(conn_, p.stmtName_.c_str(), p.paramCount_, (const char **) p.paramValues_,
                                     p.paramLengths_,
                                     p.paramFormats_, 0);
            if (!checkPQresultStatus(PQresultStatus(result_))) {
                OkPgResultSet res(PQerrorMessage(conn_));
                rollBack();
                PQclear(result_);
                promise->setValue(res);
            } else {
                OkPgResultSet res(result_);
                commit();
                PQclear(result_);
                promise->setValue(res);
            }
        } catch (const OkPgException &ex) {
            rollBack();
            OkPgResultSet res(ex.message_);
            promise->setValue(res);
        }
    });
    th.detach();
    return furture;

}